package com.atguigu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.app.func.MyAsyncFunction;
import com.atguigu.bean.ExamPaperBean;
import com.atguigu.utils.DateFormatUtil;
import com.atguigu.utils.MyClickHouseUtil;
import com.atguigu.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;


import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * ClassName: Dws05_ExamPaper01
 * Package: com.atguigu.app.dws
 * Description:
 *
 * @Author ChenJun(有志男青年)
 * @Create 2023/5/17 14:54
 * @Version 1.0
 */

//数据流: web/app -> Maxwell -> Kafka(ODS) -> FlinkApp -> Kafka(DWD) -> FlinkApp -> ClickHouse(DWS)
//程 序:  Mock -> Maxwell -> Kafka(ZK) -> BaseDBApp -> Kafka(ZK) -> Dws05ExamPaper01 -> ClickHouse(ZK)

public class Dws05Test {
    public static void main(String[] args) throws Exception {
        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
//        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(18*60+5));
        //  状态后端设置
//        env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
//        env.getCheckpointConfig().enableExternalizedCheckpoints(
//                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
//        );
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(
//                10, Time.of(1L, TimeUnit.DAYS), Time.of(3L, TimeUnit.MINUTES)
//        ));
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
//        System.setProperty("HADOOP_USER_NAME", "atguigu");
        //2.从kafka DWD层读取dwd_study_test_exam主题的数据
        DataStreamSource<String> kafkaDS = env.fromSource(MyKafkaUtil.getKafkaSource("dwd_test_exam", "exam_paper"),
                WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<String>() {
                    @Override
                    public long extractTimestamp(String element, long recordTimestamp) {
                        JSONObject jsonObject = JSON.parseObject(element);
                        return jsonObject.getLong("create_time");
                    }
                }),
                "kafka-source");
        kafkaDS.print("kafkaDS>>>>>");
        //3.转换为JavaBean
        SingleOutputStreamOperator<ExamPaperBean> examPaperDS = kafkaDS.map(new MapFunction<String, ExamPaperBean>() {
            @Override
            public ExamPaperBean map(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                return new ExamPaperBean("", "", jsonObject.getString("paper_id"), "", "", 1L, jsonObject.getBigDecimal("score"),
                        jsonObject.getLong(
                                "duration_sec"),
                        0L);
            }
        });
        //4.按照 paper_id 进行Keyby 分组 开窗 聚合
        SingleOutputStreamOperator<ExamPaperBean> reduceDS = examPaperDS.keyBy(ExamPaperBean::getPaperId).window(TumblingEventTimeWindows.of(Time.seconds(10))).reduce(new ReduceFunction<ExamPaperBean>() {
            @Override
            public ExamPaperBean reduce(ExamPaperBean value1, ExamPaperBean value2) throws Exception {
                value1.setUserNum(value1.getUserNum() + value2.getUserNum());
                value1.setScore(value1.getScore().add(value2.getScore()));
                value1.setDurationSec(value1.getDurationSec() + value2.getDurationSec());

                return value1;
            }
        }, new WindowFunction<ExamPaperBean, ExamPaperBean, String, TimeWindow>() {
            @Override
            public void apply(String s, TimeWindow window, Iterable<ExamPaperBean> input, Collector<ExamPaperBean> out) throws Exception {
                ExamPaperBean next = input.iterator().next();
                next.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                next.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                next.setTs(System.currentTimeMillis());

                out.collect(next);
            }
        });

        reduceDS.print("reduceDS>>>>");

        //5.关联维表
        SingleOutputStreamOperator<ExamPaperBean> resultDS = AsyncDataStream.unorderedWait(reduceDS, new MyAsyncFunction<ExamPaperBean>("DIM_TEST_PAPER") {
            @Override
            public void join(ExamPaperBean examPaper, JSONObject dimInfo) throws Exception {
                examPaper.setPaperTitle(dimInfo.getString("PAPER_TITLE"));
                examPaper.setCourseId(dimInfo.getString("COURSE_ID"));
            }

            @Override
            public String getKey(ExamPaperBean examPaper) throws Exception {
                return examPaper.getPaperId();
            }
        }, 60, TimeUnit.SECONDS);
        //6.写出数据
        resultDS.print("");
        resultDS.addSink(MyClickHouseUtil.getSinkFunction("insert into dws_exam_paper_paperid_window values(?,?,?,?,?,?,?,?,?)"));

        //7.执行
        env.execute();
    }
}
